package com.lagou.wwyan.dubbo.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter {
    private Map<String, List<RequestTime>> methodTime = new ConcurrentHashMap<>();
    private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

    public TPMonitorFilter() {
        //启动定时器
        executorService.scheduleWithFixedDelay(this::calculation,5,5, TimeUnit.SECONDS);
    }

    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        if (RpcContext.getContext().isAsyncStarted()){
            System.out.println("异步调用，不进行监控");
            return invoker.invoke(invocation);
        }
        String url = invoker.getUrl().toIdentityString() + "." + invocation.getMethodName();
//        System.out.println(url + "进入TPMonitorFilter");
        Result result = null;
        RequestTime requestTime = new RequestTime();
        requestTime.start();
        try {
            result = invoker.invoke(invocation);

        } finally {
            requestTime.end();
            methodTime.computeIfAbsent(url,s -> Collections.synchronizedList(new ArrayList<>())).add(requestTime);
        }
        return result;
    }

    /**
     * 统计各个接口请求时间 TP90 TP99
     */
    private void calculation(){
        System.out.println("------------开始统计接口请求-------");
        for (Map.Entry<String, List<RequestTime>> entry : methodTime.entrySet()) {
            String url = entry.getKey();
            List<RequestTime> requestTimes = entry.getValue();
            //删除超过1分钟的数据
            long timeMillis = System.currentTimeMillis();
            requestTimes.removeIf(requestTime -> timeMillis - requestTime.getStartTime()  >6000);


            //最近一分钟内 排序

            List<RequestTime> list = requestTimes.stream()
                    .sorted().collect(Collectors.toList());

            // tp 90
            RequestTime time90 = getRequestTime( list, 0.9);
            System.out.println(url+"，TP90 请求时间为"+time90.len +"ms");

            // tp 99
            RequestTime time99 = getRequestTime(list, 0.99);
            System.out.println(url+"，TP99 请求时间为"+time99.len +"ms");
        }
        System.out.println("------------结束统计接口请求-------");
    }

    /**
     * 根据比率 获取请求时间
     * @param list
     * @param rate
     * @return
     */
    private RequestTime getRequestTime(List<RequestTime> list,double rate) {
        int size = list.size();
        if (size <= 0){
            return new RequestTime();
        }
        int index = (int) Math.floor(size * rate);
        index = Math.min(index,size - 1);
        return list.get(index);
    }


    public static class RequestTime implements Comparable<RequestTime>{
        private long startTime;
        private long endTime;
        private long len;

        public RequestTime() {

        }

        public  void start(){
            startTime = System.currentTimeMillis();
        }
        public void end(){
            endTime = System.currentTimeMillis();
            len = endTime - startTime;
        }


        public long getStartTime() {
            return startTime;
        }

        public long getEndTime() {
            return endTime;
        }

        public long getLen() {
            return len;
        }

        @Override
        public int compareTo(RequestTime o) {
            return Comparator.comparing(RequestTime::getLen).compare(this,o);
        }
    }
}
